package com.atguigu.gmall.realtime.app.func;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.beans.DimJoinFunction;
import com.atguigu.gmall.realtime.utils.DimUtil;
import com.atguigu.gmall.realtime.utils.ThreadPoolUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;

import java.util.Collections;
import java.util.concurrent.ExecutorService;

/**
 * @author Felix
 * @date 2023/1/10
 * 发送异步请求进行维度关联的类
 * 模板方法设计模式
 *      在父类中定义完成某一个功能的核心算法骨架(步骤)。
 *      有些步骤的实现延迟到子类中去完成，在不改变核心算法骨架的前提下，每个子类都可以有自己不同的实现。
 */
public abstract class DimAsyncFunction<T> extends RichAsyncFunction<T,T> implements DimJoinFunction<T> {

    private ExecutorService executorService;

    private String tableName;

    public DimAsyncFunction(String tableName) {
        this.tableName = tableName;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
        //1.获取线程池对象
        executorService = ThreadPoolUtil.getInstance();
    }

    @Override
    public void asyncInvoke(T obj, ResultFuture<T> resultFuture) throws Exception {
        //2.启动线程
        executorService.submit(
            new Runnable() {
                @Override
                public void run() {
                    //3.根据流中的对象获取要关联的维度的主键
                    String key = getKey(obj);
                    //4.根据维度的主键 调用DimUtil工具类获取维度对象
                    JSONObject dimInfoJsonObj = DimUtil.getDimInfo(tableName, key);
                    if(dimInfoJsonObj != null){
                        //5.将维度对象的属性补充到流中对象上
                        join(dimInfoJsonObj,obj);
                    }
                    //6.将补充完维度属性的流中对象 向下游传递
                    resultFuture.complete(Collections.singleton(obj));
                }
            }
        );
    }
}
